package com.ferry.config.client.invoker;

import com.ferry.config.client.annotation.FerryConfig;
import com.ferry.config.client.listener.AnnotationConfigChangeListener;
import com.ferry.config.client.listener.ConfigChangeListener;
import com.ferry.config.client.listener.GlobalConfigChangeListener;
import com.ferry.config.client.util.ReflectUtil;
import com.ferry.config.client.vo.ConfigRepository;
import com.google.common.collect.MapDifference;
import com.google.common.collect.Maps;
import org.reflections.Reflections;
import org.reflections.scanners.FieldAnnotationsScanner;
import org.reflections.util.ClasspathHelper;
import org.reflections.util.ConfigurationBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

/**
 * Created by daiyong
 */
public class HttpClientInvoker extends AbstractClientInvoker implements ClientInvoker {

	private static final Logger LOGGER = LoggerFactory.getLogger(HttpClientInvoker.class);

	private ConcurrentMap<String, Set<ConfigChangeListener>> innerListeners = new ConcurrentHashMap<String, Set<ConfigChangeListener>>();

	private ConcurrentMap<String, Set<ConfigChangeListener>> customListeners = new ConcurrentHashMap<String, Set<ConfigChangeListener>>();

	private volatile GlobalConfigChangeListener globalConfigChangeListener = new GlobalConfigChangeListener() {
		@Override
		public void changed(Map<String, String> oldData, Map<String, String> newData) {
			//ignore
		}
	};

	private ExecutorService configListenerExecutor = Executors.newFixedThreadPool(1);

	private ExecutorService globalConfigListenerExecutor = Executors.newFixedThreadPool(1);

	//拉取线程
	private static ScheduledExecutorService getExecutorService = Executors.newScheduledThreadPool(1, new ThreadFactory() {
				public Thread newThread(Runnable r) {
					Thread t = Executors.defaultThreadFactory().newThread(r);
					if (t != null) {
						t.setDaemon(true);
					}
					return t;
				}
			}
	);

	private ConfigRepository configRepository;

	private volatile Reflections reflections;
	private static final Object lock = new Object();
	private static final String SCAN_BASE_PACKAGE = ".";

	private static final Integer PULL_PERIOD = 500; //ms

	private String nameSpace;

	public HttpClientInvoker(){}

	@Override
	public void init() {
		initCacheConfig();

		initScanAnnotation();

		getExecutorService.scheduleWithFixedDelay(new Runnable() {
			@Override
			public void run() {
				try {
					getConfig();
				} catch (Exception e) {
					LOGGER.error("pullExecutorService.scheduleWithFixedDelay Exception", e);
				}
			}
		}, PULL_PERIOD, PULL_PERIOD, TimeUnit.MILLISECONDS);
	}

	private void initCacheConfig() {
		configRepository = testGetConfig();
	}

	@Override
	public void destroy() {

	}

	private void addInnerListener(String key, ConfigChangeListener listener) {
		if (!innerListeners.containsKey(key)) {
			synchronized (lock) {
				if (!innerListeners.containsKey(key)) {
					innerListeners.put(key, new HashSet<ConfigChangeListener>());
				}
			}
		}
		innerListeners.get(key).add(listener);
	}

	@Override
	public void addListener(String key, ConfigChangeListener listener) {
		if (!customListeners.containsKey(key)) {
			synchronized (lock) {
				if (!customListeners.containsKey(key)) {
					customListeners.put(key, new HashSet<ConfigChangeListener>());
				}
			}
		}
		customListeners.get(key).add(listener);
	}

	@Override
	public void removeListener(String key, ConfigChangeListener listener) {
		if (customListeners.containsKey(key)) {
			synchronized (lock) {
				if (customListeners.containsKey(key)) {
					Set<ConfigChangeListener> listeners = customListeners.get(key);
					if (null != listeners && !listeners.isEmpty()) {
						customListeners.get(key).remove(listener);
					}
				}
			}
		}
	}

	@Override
	public String getValue(String key) {
		return configRepository.getConfig().get(key);
	}

	@Override
	public Map<String, String> getAllKeyValues() {
		return new HashMap<String, String>(configRepository.getConfig());
	}

	@Override
	public Set<String> getAllKeys() {
		return new HashSet<String>(configRepository.getConfig().keySet());
	}

	@Override
	public void setGlobalConfigChangeListener(GlobalConfigChangeListener globalConfigChangeListener) {
		this.globalConfigChangeListener = globalConfigChangeListener;
	}

	@Override
	protected void scanAnnotation() {
		initReflections();

		Set<Field> fields = reflections.getFieldsAnnotatedWith(FerryConfig.class);
		Set<String> configKeys = new HashSet<String>();
		Map<String, String> newConfig = Maps.newHashMap();

		for (Field f : fields) {
			if (f.isAnnotationPresent(FerryConfig.class)) {
				FerryConfig configParam = f.getAnnotation(FerryConfig.class);
				String key = configParam.key();
				configKeys.add(key);
				watchField(f, key, newConfig);
			}
		}

	}

	/**
	 * 初始化反射器
	 */
	private void initReflections() {
		if (null == reflections) {
			synchronized (lock) {
				if (null == reflections) {
					reflections = new Reflections(new ConfigurationBuilder()
							.setUrls(ClasspathHelper.forPackage(SCAN_BASE_PACKAGE))
							.setScanners(new FieldAnnotationsScanner()));
				}
			}
		}
	}

	private void watchField(final Field field, String key, Map<String, String> newConfig) {
		try {
			Class fieldType = field.getDeclaringClass();
			addInnerListener(key, new AnnotationConfigChangeListener(fieldType));

			ConcurrentMap<String, String> config = configRepository.getConfig();
			field.setAccessible(Boolean.TRUE);
			if (config.keySet().contains(key)) {
				Object value = ReflectUtil.transferValueType(field, config.get(key));
				field.set(fieldType, value);
			} else {
				Object defaultValue = field.get(fieldType);
				if (null != defaultValue) {
					String value = defaultValue.toString();
					configRepository.getConfig().put(key, value);
					newConfig.put(key, value);
				}
			}
		} catch (Exception e) {
			LOGGER.error("watchField Exception, fieldName:{}", field.getName(), e);
		}
	}

	private void getConfig() {
		ConfigRepository newConfigRepository = testGetConfig();
		try {

			// 更新缓存版本
			final ConcurrentMap<String, String> oldData = configRepository.getConfig();
			final ConcurrentMap<String, String> newData = newConfigRepository.getConfig();
			// 更新缓存
			configRepository = newConfigRepository;
			// 对比data
			diff(oldData, newData);

			globalConfigListenerExecutor.execute(new Runnable() {
				@Override
				public void run() {
					globalConfigChangeListener.changed(oldData, newData);
				}
			});

		} catch (Exception e) {
			LOGGER.error("pull Exception", e);
		}
	}

	private void diff(final Map<String, String> oldData, final Map<String, String> newData) {
		if (oldData == null || newData == null) {
			return;
		}
		MapDifference<String, String> diffData = Maps.difference(oldData, newData);
		configDeleted(diffData);
		configUpdated(diffData);
		configAdded(diffData);
	}

	private void configAdded(MapDifference<String, String> diffData) {
		for (Map.Entry<String, String> addedEntry : diffData.entriesOnlyOnRight().entrySet()) {
			configChanged(addedEntry.getKey(), null, addedEntry.getValue());
		}
	}

	private void configUpdated(MapDifference<String, String> diffData) {
		for (Map.Entry<String, MapDifference.ValueDifference<String>> changedEntry : diffData.entriesDiffering().entrySet()) {
			MapDifference.ValueDifference<String> oldAndNewValue = changedEntry.getValue();
			configChanged(changedEntry.getKey(), oldAndNewValue.leftValue(), oldAndNewValue.rightValue());
		}
	}

	private void configDeleted(MapDifference<String, String> diffData) {
		for (Map.Entry<String, String> deletedEntry : diffData.entriesOnlyOnLeft().entrySet()) {
			configChanged(deletedEntry.getKey(), deletedEntry.getValue(), null);
		}
	}

	private void configChanged(final String key, final String oldValue, final String newValue) {
		configListenerExecutor.execute(new Runnable() {
			@Override
			public void run() {
				if (innerListeners.containsKey(key)) {
					for (final ConfigChangeListener listener : innerListeners.get(key)) {
						listener.changed(key, oldValue, newValue);
					}
				}
				if (customListeners.containsKey(key)) {
					executeCustomListener(key, oldValue, newValue);
				}
			}
		});
	}

	private void executeCustomListener(final String key, final String oldValue, final String newValue) {
		Set<ConfigChangeListener> configChangeListeners = customListeners.get(key);
		if (!CollectionUtils.isEmpty(configChangeListeners)) {
			for (final ConfigChangeListener listener : configChangeListeners) {
				listener.changed(key, oldValue, newValue);
			}
		}
	}

	public void setNameSpace(String nameSpace) {
		this.nameSpace = nameSpace;
	}

	private ConfigRepository testGetConfig() {
		ConfigRepository repo = new ConfigRepository();

		ConcurrentMap<String, String> map = Maps.newConcurrentMap();
		map.put("name", "daiyong");
		map.put("age", "1");
		repo.setConfig(map);
		return repo;

	}
}
